package com.lagou.mall.esflow.server.config;

import com.alibaba.fastjson.JSON;
import com.lagou.mall.esflow.client.BulkUpsert;
import com.lagou.mall.esflow.server.ESFlowSystemException;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.DocWriteResponse.Result;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ElasticSearchTemplate {

    @Value("${elasticsearch.ip}")
    private String ip;
    @Value("${elasticsearch.port}")
    private int port;
    @Value("${elasticsearch.user}")
    private String user;
    @Value("${elasticsearch.password}")
    private String password;
    private RestHighLevelClient client;
    @Value("${elasticsearch.connectTimeout}")
    private int connectTimeout;
    @Value("${elasticsearch.socketTimeout}")
    private int socketTimeout;
    @Value("${elasticsearch.connectionRequestTimeout}")
    private int connectionRequestTimeout;
    @Value("${elasticsearch.maxConnTotal}")
    private int maxConnTotal;
    @Value("${elasticsearch.maxConnPerRoute}")
    private int maxConnPerRoute;
    @Value("${elasticsearch.maxRetryTimeoutMillis}")
    private int maxRetryTimeoutMillis;


    public static final String DOCUMENT_TYPE = "_doc";
    private static final String LANG = "painless";

    /**
     * 初始化RestHighLevelClient
     */
    @PostConstruct
    public void init() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
        //配置化
        RestClientBuilder builder =
                RestClient.builder(new HttpHost(ip, port, "http")).setRequestConfigCallback(requestConfigBuilder -> {
                    requestConfigBuilder.setConnectTimeout(connectTimeout);
                    requestConfigBuilder.setSocketTimeout(socketTimeout);
                    requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeout);
                    return requestConfigBuilder;
                }).setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.setMaxConnTotal(maxConnTotal);
                    httpClientBuilder.setMaxConnPerRoute(maxConnPerRoute);
                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    return httpClientBuilder;
                });

        client = new RestHighLevelClient(builder);
        log.info("RestHighLevelClient  init success...");
    }

    public RestHighLevelClient getClient() {
        return client;
    }


    @PreDestroy
    public void destroy() throws IOException {
        if (null != client) {
            client.close();
        }
    }

    public boolean delete(String id, String index) {
        DeleteRequest request = new DeleteRequest(index, DOCUMENT_TYPE, id);
        DeleteResponse response = null;
        try {
            response = client.delete(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException exception) {
            if (exception.status() == RestStatus.CONFLICT) {
                log.error("delete error conflict {},{}", index, id);
            }
            throw ESFlowSystemException.error("restHighLevelClient delete exception", exception);
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient delete exception", e);
        }
        if (Objects.requireNonNull(response).getResult() == DocWriteResponse.Result.NOT_FOUND) {
            log.warn("delete error not_found {},{}", index, id);
            // 文档已删除只打印日志，不删除
        }
        return true;
    }

    public boolean upsert(String id, String index, String body) {
        return upsert(id, index, DOCUMENT_TYPE, body);
    }

    public boolean upsert(String id, String index, String documentType, String body) {
        UpdateRequest request = new UpdateRequest(index, documentType, id);
        request.upsert(body, XContentType.JSON)
                .doc(body, XContentType.JSON)
                .docAsUpsert(true)
                .waitForActiveShards(0);
        try {
            UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
            if (updateResponse.getResult() == DocWriteResponse.Result.CREATED ||
                    updateResponse.getResult() == DocWriteResponse.Result.UPDATED ||
                    updateResponse.getResult() == Result.NOOP) {
                return true;
            }
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient upsert exception", e);
        }
        return false;
    }


    public void bulkUpsert(String index, Collection<BulkUpsert> bulkUpserts) {
        if (CollectionUtils.isEmpty(bulkUpserts)) {
            return;
        }
        BulkRequest bulkRequest = new BulkRequest();
        bulkUpserts.stream().map(
                bulkUpsert ->
                        new UpdateRequest(index, DOCUMENT_TYPE, bulkUpsert.getId())
                                .upsert(bulkUpsert.getBody(), XContentType.JSON)
                                .doc(bulkUpsert.getBody(), XContentType.JSON)
                                .docAsUpsert(true)
        ).forEach(bulkRequest::add);
        try {
            bulkRequest.waitForActiveShards(0);
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                throw ESFlowSystemException.error(bulkResponse.buildFailureMessage(), new Throwable());
            }
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient bulkUpdate exception", e);
        }
    }

    /**
     * 大批量数据 分批处理
     *
     * @param size 每批数据大小
     * @param retryTimes 每批数据失败后可重试次数
     */
    public void bulkUpsert(String index, List<BulkUpsert> bulkUpserts, int size, int retryTimes) {
        for (int i = 0, total = bulkUpserts.size(); i < total; ) {
            int to = i + size;
            if (to > total) {
                to = total;
            }
            reTryBatch(index, bulkUpserts.subList(i, to), retryTimes);
            i = to;
        }
    }

    private void reTryBatch(String index, List<BulkUpsert> bulkUpserts, int time) {
        try {
            this.bulkUpsert(index, bulkUpserts);
        } catch (Exception e) {
            if (time == 0) {
                log.error("reTryBatch too many times {}", JSON.toJSONString(bulkUpserts));
                throw e;
            }
            reTryBatch(index, bulkUpserts, --time);
        }
    }

    /**
     * @param index 索引名
     * @param id 索引id
     * @param scriptId 脚本
     */
    public void updateScript(String id, String index, String scriptId, Map<String, Object> params) {
        Script script = new Script(ScriptType.INLINE, LANG, scriptId, params);
        UpdateRequest updateRequest = new UpdateRequest(index, DOCUMENT_TYPE, id)
                .script(script)
                .retryOnConflict(3);
        try {
            client.update(updateRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient update exception", e);
        }
    }

    public void bulkUpdateScript(String index, String scriptId, Collection<BulkUpsert> bulkUpdates) {
        if (CollectionUtils.isEmpty(bulkUpdates)) {
            return;
        }
        BulkRequest bulkRequest = new BulkRequest();
        bulkUpdates.stream().map(bulkUpsert -> new UpdateRequest(index, DOCUMENT_TYPE, bulkUpsert.getId())
                        .script(new Script(ScriptType.INLINE, LANG, scriptId, bulkUpsert.getParams()))
                        .retryOnConflict(3))
                .forEach(bulkRequest::add);
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                throw ESFlowSystemException.error(bulkResponse.buildFailureMessage(), new Throwable());
            }
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient bulkUpdate exception", e);
        }
    }


    public void bulkUpdateScript(String index, String scriptId, List<BulkUpsert> bulkUpdates, int size, int retryTimes) {
        for (int i = 0, total = bulkUpdates.size(); i < total; ) {
            int to = i + size;
            if (to > total) {
                to = total;
            }
            reTryBatchBulkUpdateScript(index, scriptId, bulkUpdates.subList(i, to), retryTimes);
            i = to;
        }
    }

    private void reTryBatchBulkUpdateScript(String index, String scriptId, List<BulkUpsert> bulkUpserts, int time) {
        try {
            this.bulkUpdateScript(index, scriptId, bulkUpserts);
        } catch (Exception e) {
            if (time == 0) {
                log.error("reTryBatchBulkUpdateScript too many times {}", JSON.toJSONString(bulkUpserts));
                throw e;
            }
            reTryBatch(index, bulkUpserts, --time);
        }
    }

    public void insertOnly(String id, String index, String body) {
        IndexRequest request = new IndexRequest(index, DOCUMENT_TYPE, id)
                .create(true)
                .source(body, XContentType.JSON);
        try {
            client.index(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                log.warn("insertOnly error already_exist {},{}", index, id);
                // 文档已存在，只打印日志，不抛出异常
            } else {
                throw ESFlowSystemException.error("restHighLevelClient insertOnly exception", e);
            }
        } catch (Exception e) {
            throw ESFlowSystemException.error("restHighLevelClient insertOnly exception", e);
        }
    }

    public boolean createIndexWithWholeSource(String source, String index) {
        CreateIndexRequest request = new CreateIndexRequest(index);
        request.source(source, XContentType.JSON);
        try {
            CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
            return createIndexResponse.isAcknowledged();
        } catch (IOException e) {
            throw ESFlowSystemException.error("restHighLevelClient createIndexWithWholeSource exception", e);
        }
    }


    public boolean updateIndex(String indexName, String fieldName, String type) {
        PutMappingRequest request = new PutMappingRequest("twitter");
        Map<String, Object> jsonMap = new HashMap<>();
        Map<String, Object> message = new HashMap<>();
        Map<String, Object> properties = new HashMap<>();
        message.put("type", type);
        properties.put("message", message);
        jsonMap.put("properties", properties);
        request.source(jsonMap);

        try {
            AcknowledgedResponse response = client.indices().putMapping(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
            throw ESFlowSystemException.error("putMapping counter error", e);
        }

    }

}
